package workflow

import (
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/asynkron/protoactor-go/actor"
	"github.com/google/uuid"
	"github.com/louloulin/dataflare/pkg/actor/system"
	"github.com/louloulin/dataflare/pkg/workflow/processor"
)

// ProcessorActor is an actor that processes data
type ProcessorActor struct {
	// id is the processor ID
	id string
	// definition is the processor definition
	definition ProcessorDefinition
	// instanceID is the workflow instance ID
	instanceID string
	// stateManager is the state manager PID
	stateManager *actor.PID
	// metricsCollector is the metrics collector PID
	metricsCollector *actor.PID
	// processor is the data processor
	processor processor.Processor
	// targets are the target nodes
	targets map[string]*actor.PID
	// status is the node status
	status string
	// mu is a mutex for protecting the targets and status
	mu sync.RWMutex
	// logger is the processor actor logger
	logger *log.Logger
	// running indicates if the processor is running
	running bool
	// processedCount is the number of records processed
	processedCount int64
	// errorCount is the number of errors
	errorCount int64
	// lastError is the last error
	lastError error
	// lastProcessTime is the last time data was processed
	lastProcessTime time.Time
	// sourceActors are the source actors that send data to this processor
	sourceActors map[string]*actor.PID
}

// NewProcessorActor creates a new processor actor
func NewProcessorActor(id string, definition ProcessorDefinition, instanceID string, stateManager *actor.PID, metricsCollector *actor.PID) actor.Actor {
	return &ProcessorActor{
		id:               id,
		definition:       definition,
		instanceID:       instanceID,
		stateManager:     stateManager,
		metricsCollector: metricsCollector,
		targets:          make(map[string]*actor.PID),
		sourceActors:     make(map[string]*actor.PID),
		status:           "created",
		logger:           log.New(log.Writer(), fmt.Sprintf("[ProcessorActor-%s-%s] ", instanceID, id), log.LstdFlags),
	}
}

// Receive handles messages for the processor actor
func (p *ProcessorActor) Receive(context actor.Context) {
	switch msg := context.Message().(type) {
	case *actor.Started:
		p.handleStarted(context)
	case *actor.Stopping:
		p.handleStopping(context)
	case *ConnectNodeCommand:
		p.handleConnectNode(context, msg)
	case *StartNodeCommand:
		p.handleStartNode(context, msg)
	case *StopNodeCommand:
		p.handleStopNode(context)
	case *GetNodeStatusCommand:
		p.handleGetNodeStatus(context)
	case *system.DataMessage:
		p.handleDataMessage(context, msg)
	default:
		p.logger.Printf("Received unknown message: %T", msg)
	}
}

// handleStarted handles the Started message
func (p *ProcessorActor) handleStarted(context actor.Context) {
	p.logger.Println("Processor actor started")

	// Create processor
	var err error
	p.processor, err = p.createProcessor()
	if err != nil {
		p.logger.Printf("Failed to create processor: %v", err)
		p.status = "failed"
		context.Send(context.Parent(), &NodeErrorMessage{
			NodeID: p.id,
			Error:  fmt.Sprintf("Failed to create processor: %v", err),
		})
		return
	}

	// Report metrics
	if p.metricsCollector != nil {
		context.Send(p.metricsCollector, &system.ReportMetricCommand{
			Name:  "processor.started",
			Value: 1,
			Tags: map[string]string{
				"processor_id": p.id,
				"instance_id":  p.instanceID,
				"type":         p.definition.Type,
			},
		})
	}

	p.status = "initialized"
	p.logger.Println("Processor actor initialized")
}

// handleStopping handles the Stopping message
func (p *ProcessorActor) handleStopping(context actor.Context) {
	p.logger.Println("Processor actor stopping")

	// Stop processor
	if p.processor != nil {
		err := p.processor.Close()
		if err != nil {
			p.logger.Printf("Failed to close processor: %v", err)
		}
	}

	// Report metrics
	if p.metricsCollector != nil {
		context.Send(p.metricsCollector, &system.ReportMetricCommand{
			Name:  "processor.stopped",
			Value: 1,
			Tags: map[string]string{
				"processor_id": p.id,
				"instance_id":  p.instanceID,
				"type":         p.definition.Type,
			},
		})
	}

	p.status = "stopped"
}

// handleConnectNode handles the ConnectNodeCommand message
func (p *ProcessorActor) handleConnectNode(context actor.Context, msg *ConnectNodeCommand) {
	p.mu.Lock()
	defer p.mu.Unlock()

	// Add target
	p.targets[msg.TargetID] = msg.TargetPID

	// Respond with success
	context.Respond(&ConnectNodeResponse{
		Success: true,
	})

	p.logger.Printf("Connected to target node: %s", msg.TargetID)
}

// handleStartNode handles the StartNodeCommand message
func (p *ProcessorActor) handleStartNode(context actor.Context, msg *StartNodeCommand) {
	p.mu.Lock()
	defer p.mu.Unlock()

	// Check if already running
	if p.running {
		context.Respond(&StartNodeResponse{
			Success: false,
			Error:   "processor already running",
		})
		return
	}

	// Initialize processor
	err := p.processor.Init(p.definition.Config)
	if err != nil {
		p.logger.Printf("Failed to initialize processor: %v", err)
		context.Respond(&StartNodeResponse{
			Success: false,
			Error:   fmt.Sprintf("failed to initialize processor: %v", err),
		})
		return
	}

	// Update state
	p.running = true
	p.status = "running"

	// Respond with success
	context.Respond(&StartNodeResponse{
		Success: true,
	})

	// Report status change
	context.Send(context.Parent(), &NodeStatusChangedMessage{
		NodeID: p.id,
		Status: "running",
	})

	p.logger.Println("Processor started")
}

// handleStopNode handles the StopNodeCommand message
func (p *ProcessorActor) handleStopNode(context actor.Context) {
	p.mu.Lock()
	defer p.mu.Unlock()

	// Check if running
	if !p.running {
		context.Respond(&StopNodeResponse{
			Success: false,
			Error:   "processor not running",
		})
		return
	}

	// Update state
	p.running = false
	p.status = "stopped"

	// Respond with success
	context.Respond(&StopNodeResponse{
		Success: true,
	})

	// Report status change
	context.Send(context.Parent(), &NodeStatusChangedMessage{
		NodeID: p.id,
		Status: "stopped",
	})

	p.logger.Println("Processor stopped")
}

// handleGetNodeStatus handles the GetNodeStatusCommand message
func (p *ProcessorActor) handleGetNodeStatus(context actor.Context) {
	p.mu.RLock()
	defer p.mu.RUnlock()

	// Create metrics
	metrics := map[string]interface{}{
		"processed_count":  p.processedCount,
		"error_count":      p.errorCount,
		"last_process_time": p.lastProcessTime,
	}

	// Respond with status
	context.Respond(&GetNodeStatusResponse{
		Status:  p.status,
		Metrics: metrics,
		Error:   p.lastError,
	})
}

// handleDataMessage handles the DataMessage message
func (p *ProcessorActor) handleDataMessage(context actor.Context, msg *system.DataMessage) {
	p.mu.RLock()
	running := p.running
	p.mu.RUnlock()

	// Check if running
	if !running {
		p.logger.Println("Received data message while not running, ignoring")
		return
	}

	// Process data
	startTime := time.Now()
	result, err := p.processor.Process(msg.Data, msg.Metadata)
	processingTime := time.Since(startTime)

	// Update state
	p.mu.Lock()
	p.lastProcessTime = time.Now()
	p.mu.Unlock()

	// Handle error
	if err != nil {
		p.logger.Printf("Failed to process data: %v", err)
		p.mu.Lock()
		p.lastError = err
		p.errorCount++
		p.mu.Unlock()

		// Report error metrics
		if p.metricsCollector != nil {
			context.Send(p.metricsCollector, &system.ReportMetricCommand{
				Name:  "processor.errors",
				Value: 1,
				Tags: map[string]string{
					"processor_id": p.id,
					"instance_id":  p.instanceID,
					"type":         p.definition.Type,
				},
			})
		}

		// Return credit to source
		if sourceID, ok := msg.Metadata["source_id"]; ok {
			if sourcePID, ok := p.sourceActors[sourceID]; ok {
				context.Send(sourcePID, &CreditReturnedMessage{})
			}
		}

		return
	}

	// Update processed count
	p.mu.Lock()
	p.processedCount++
	p.mu.Unlock()

	// Report metrics
	if p.metricsCollector != nil {
		context.Send(p.metricsCollector, &system.ReportMetricCommand{
			Name:  "processor.records_processed",
			Value: 1,
			Tags: map[string]string{
				"processor_id": p.id,
				"instance_id":  p.instanceID,
				"type":         p.definition.Type,
			},
		})

		context.Send(p.metricsCollector, &system.ReportMetricCommand{
			Name:  "processor.processing_time",
			Value: float64(processingTime.Milliseconds()),
			Tags: map[string]string{
				"processor_id": p.id,
				"instance_id":  p.instanceID,
				"type":         p.definition.Type,
			},
		})
	}

	// Send result to targets
	p.sendResultToTargets(context, result, msg.Metadata)

	// Return credit to source
	if sourceID, ok := msg.Metadata["source_id"]; ok {
		if sourcePID, ok := p.sourceActors[sourceID]; ok {
			context.Send(sourcePID, &CreditReturnedMessage{})
		}
	}
}

// sendResultToTargets sends the result to target nodes
func (p *ProcessorActor) sendResultToTargets(context actor.Context, result []byte, metadata map[string]string) {
	p.mu.RLock()
	defer p.mu.RUnlock()

	// Send result to each target
	for targetID, targetPID := range p.targets {
		// Create data message
		message := &system.DataMessage{
			MessageID: uuid.New().String(),
			Data:      result,
			Metadata: map[string]string{
				"processor_id": p.id,
				"instance_id":  p.instanceID,
			},
			Timestamp:     time.Now(),
			SourceID:      p.id,
			DestinationID: targetID,
		}

		// Copy original metadata
		for k, v := range metadata {
			message.Metadata[k] = v
		}

		// Send message
		context.Send(targetPID, message)
	}
}

// createProcessor creates a data processor
func (p *ProcessorActor) createProcessor() (processor.Processor, error) {
	// Get processor factory
	factory := processor.GetProcessorFactory(p.definition.Type)
	if factory == nil {
		return nil, fmt.Errorf("unknown processor type: %s", p.definition.Type)
	}

	// Create processor
	return factory()
}
